package com.raymond.queue.client.service;

import com.raymond.queue.client.QueueClient;
import com.raymond.queue.client.msg.MessageManages;
import com.raymond.queue.common.enums.GrowMode;
import com.raymond.queue.common.msg.request.ConsumptionMsgRequest;
import com.raymond.queue.common.msg.request.CreateConsumptionRequest;
import com.raymond.queue.common.msg.response.ConsumptionMsgResponse;
import com.raymond.queue.common.msg.response.CreateConsumptionResponse;
import com.raymond.queue.common.utils.SequenceIdUtil;
import io.netty.util.internal.StringUtil;

/**
 * 消费者
 *
 * @author :  raymond
 * @version :  V1.0
 * @date :  2021-07-19 11:39
 */
public class Consumption {
    private QueueClient queueClient;


    public Consumption(QueueClient queueClient) {
        this.queueClient = queueClient;
    }

    /**
     * 拉取一条消息
     * @param topic topic
     * @param groupName 消费组
     * @return 响应结果
     */
    public ConsumptionMsgResponse poll(String topic, String groupName) {
        return poll(topic, groupName, 1);
    }
    /**
     * 拉取多条消息
     * @param topic topic
     * @param groupName 消费组
     * @param count 条数
     * @return 响应结果
     */
    public ConsumptionMsgResponse poll(String topic, String groupName, int count) {
        ConsumptionMsgRequest request = new ConsumptionMsgRequest(SequenceIdUtil.getSequenceId(), topic, groupName, count);
        MessageManages messageManages = queueClient.syncSend(request);
        if (messageManages.isDone()) {
            return (ConsumptionMsgResponse)messageManages.getResp();
        }
        return new ConsumptionMsgResponse(0, -2);
    }

    /**
     * 异步拉取一条消息
     * @param topic topic
     * @param groupName 消费组
     * @param callback 回调接口
     */
    public void asyncPoll(String topic, String groupName, Callback callback) {
        asyncPoll(topic, groupName, callback, 1);
    }

    /**
     * 异步拉取多条消息
     * @param topic topic
     * @param groupName 消费组
     * @param callback 回调接口
     */
    public void asyncPoll(String topic, String groupName, Callback callback, int count) {
        ConsumptionMsgRequest request = new ConsumptionMsgRequest(SequenceIdUtil.getSequenceId(), topic, groupName, count);
        queueClient.asyncSend(request, callback);
    }

    /**
     * 创建一个新的消费组
     * @param topic topic
     * @param groupName 消费组名称
     * @param growMode 创建消费组的模式
     *                 0:从上次消费的开始,如果不存在上次消费,则按最后一条开始
     *                 1:从当前最后一条开始
     *                 2:从某个消费组开始
     *                 3:从首条开始
     * @param srcGroupName 来源消费组
     * @param isContinue 是否继续
     */
    public CreateConsumptionResponse createConsumption(String topic, String groupName, GrowMode growMode, String srcGroupName, boolean isContinue) {
        return createConsumption(new CreateConsumptionRequest(topic, groupName, growMode, srcGroupName, true));
    }

    /**
     * 复制消费组
     * 消费组存在的话继续上次消费,消费组不存在的话,复制来源消费组
     * @param topic 主题
     * @param groupName 需要创建的消费组
     * @param srcGroupName 来源消费组
     */
    public CreateConsumptionResponse copyGroup(String topic, String groupName, String srcGroupName) {
        return createConsumption(new CreateConsumptionRequest(topic, groupName, GrowMode.COPY_GROUP, srcGroupName, true));
    }



    /**
     * 复制消费组
     * @param topic 主题
     * @param groupName 需要创建的消费组
     * @param srcGroupName 来源消费组
     * @param isContinue
     * true:消费组存在的话继续上次消费,消费组不存在的话,复制来源消费组
     * false:不管消费组是否存在,都复制来源消费组
     * @throws Exception 异常
     */
    public CreateConsumptionResponse copyGroup(String topic, String groupName, String srcGroupName, boolean isContinue) throws Exception {
        return createConsumption(new CreateConsumptionRequest(topic, groupName, GrowMode.COPY_GROUP, srcGroupName, isContinue));
    }

    public CreateConsumptionResponse createConsumption(CreateConsumptionRequest request) {
        if (!checkCreateParam(request)) {
            return new CreateConsumptionResponse(-3, 0);
        }
        request.setSequenceId(SequenceIdUtil.getSequenceId());
        MessageManages messageManages = queueClient.syncSend(request);
        if (messageManages.isDone()) {
            return (CreateConsumptionResponse) messageManages.getResp();
        }
        return new CreateConsumptionResponse(-2, 0);
    }

    private boolean checkCreateParam(CreateConsumptionRequest request) {
        if (request == null ) {
            return false;
        }
        if (StringUtil.isNullOrEmpty(request.getTopic()) || StringUtil.isNullOrEmpty(request.getGroupName())) {
            return false;
        }
        return GrowMode.COPY_GROUP != request.getGrowMode() || !StringUtil.isNullOrEmpty(request.getSrcGroupName());
//        GrowMode.COPY_GROUP != request.getGrowMode() || !StringUtil.isNullOrEmpty(request.getSrcGroupName())
    }
}
